//package kafka.client.producer;
//
//import java.io.IOException;
//import java.io.InputStream;
//import java.util.Properties;
//import java.util.concurrent.atomic.AtomicLong;
//
//import kafka.client.entity.ResumePost;
//import kafka.client.util.JsonUtils;
//import kafka.client.util.SMS;
//import kafka.javaapi.producer.Producer;
//import kafka.producer.KeyedMessage;
//import kafka.producer.ProducerConfig;
//
//import org.apache.commons.lang3.StringUtils;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import com.bj58.spat.esbclient.ESBClient;
//import com.bj58.spat.esbclient.ESBMessage;
//import com.bj58.spat.esbclient.ESBReceiveHandler;
//import com.bj58.spat.esbclient.ESBSubject;
//
////async方式单Producer
//public class IMCProducer {
//	private static final Logger logger = LoggerFactory.getLogger(IMCProducer.class);
//	private Producer<String,String> producer;
//	private AtomicLong counter = new AtomicLong(0); 
//	private ESBClient client;
//	private String TOPIC;
//	private int numOfPartitions;
//	
//	private IMCProducer(){
//		initProducer();
//		registerExitEvent();
//	}
//	
//	private static class SingletonHolder {  
//		private static final IMCProducer INSTANCE = new IMCProducer();  
//	} 
//	
//	public static final IMCProducer getInstance() {  
//		 return SingletonHolder.INSTANCE;  
//	} 
//	
//	/*
//	 * 加载配置文件，初始化Producer
//	 */
//	private  void initProducer() {
//		try {
//			InputStream inStream = IMCProducer.class.getResourceAsStream("/kafka-producer-conf.properties");
//			Properties props = new Properties();
//			props.load(inStream);
//			this.TOPIC = props.getProperty("topic");
//			this.numOfPartitions = Integer.valueOf(props.getProperty("partition.num"));
//			props.remove("topic");
//			props.remove("partition.num");
//			ProducerConfig config = new ProducerConfig(props);
//			producer = new Producer<String,String>(config);
//		} catch (IOException e) {
//			logger.error("ERROR Occurs when init the IMCProducer: \n",e);
//			System.exit(1);
//		}
//	}
//
//	/*
//	 * 添加回调方法，关闭producer
//	 */
//    private  void registerExitEvent()
//    {
//        Runtime.getRuntime().addShutdownHook(new Thread() {
//	            public void run() {
//	            	if(producer !=null){
//	            		producer.close();
//	            	}
//	            	if(client != null){
//	            		client.closeAllChannel();
//	            	}
//	            }
//        	}
//        );
//    }
//    
//	/*
//	 * 加载配置文件，初始化 ESBClient
//	 */
//	private  void initESBClient() {
//		try {
//			InputStream inStream = IMCProducer.class.getResourceAsStream("/esb-conf.properties");
//			Properties props = new Properties();
//			props.load(inStream);
//			inStream.close();
//			client = new ESBClient(props.getProperty("url"));
//			client.setReceiveSubject(new ESBSubject(Integer.parseInt(props.getProperty("subjectId"))));
//			client.setReceiveHandler(new ESBReceiveHandler() {
//				@Override
//				public void messageReceived(ESBMessage msg) {
//		            String body = new String(msg.getBody());
//		            ResumePost resumePost = JsonUtils.fromJSON(body, ResumePost.class);
//		            if(resumePost !=null && resumePost.getOper() == 1){
//		            	long num = counter.getAndIncrement();
//			            KeyedMessage<String, String> data = new KeyedMessage<String, String>(TOPIC,String.valueOf(num % numOfPartitions), resumePost.getEntity());
//			            producer.send(data);
//		            }
//		            logger.info(body);
//				}
//			});
//			// 保持监听状态，让ESB主线程不消失
//			while (true) {
//				Thread.sleep(1000 * 60 * 60);
//			}
//		} catch (Exception e) {
//			SMS.defaultSend("IMC-Kafka异常退出!请及时查看重启!");
//			logger.error("ERROR Occurs : ",e);
//		}finally{
//			System.exit(1);
//		}
//	}
//		
//	protected void start() {
//		try {
//			if(StringUtils.isBlank(TOPIC)){
//				logger.error("TOPIC is null.");
//				System.exit(1);
//			}
//			
//			initESBClient();
//			
////	        int messageNo = 1;
////	        final int COUNT = 100;
////	 
////	        while (messageNo <= COUNT) {
////	            String body = "{\"entity\":\"{\\\"advantages2\\\":\\\"\\\",\\\"advantages3\\\":\\\"\\\",\\\"areaID\\\":1196,\\\"auditState\\\":0,\\\"birthDay\\\":799171200000,\\\"bizstate\\\":0,\\\"complete\\\":60,\\\"eucation\\\":1,\\\"exid\\\":0,\\\"exstate\\\":0,\\\"extid\\\":0,\\\"freetime\\\":\\\"\\\",\\\"gender\\\":2,\\\"inputParams\\\":{\\\"AUTHTIME\\\":\\\"1438225172000\\\"},\\\"inputPersonize\\\":{\\\"lastExpDesc\\\":\\\"3年|1|中兴（环卫工）\\\"},\\\"jobState\\\":0,\\\"lowerResumeFlag\\\":0,\\\"mobileAuth\\\":1,\\\"nativeCity\\\":0,\\\"nowPosition\\\":\\\"环卫工\\\",\\\"openIdThird\\\":0,\\\"openState\\\":1,\\\"rdoIdentity\\\":0,\\\"relateID\\\":0,\\\"resumeName\\\":\\\"求职环卫\\\",\\\"resumeid\\\":85431408681731,\\\"salary\\\":9,\\\"searchstate\\\":1,\\\"source\\\":1,\\\"targetAreaID\\\":\\\"1143\\\",\\\"targetCateID\\\":\\\"2425\\\",\\\"targetPosition\\\":\\\"求职环卫\\\",\\\"trueName\\\":\\\"测试\\\",\\\"userIP\\\":\\\"127.0.0.1\\\",\\\"userid\\\":23149986458374,\\\"workType\\\":0,\\\"workedYears\\\":0}\",\"id\":85431408681731,\"oper\":1,\"step\":1,\"whichMethod\":\"repoResume\"}";
////	            ResumePost resumePost = JsonUtils.fromJSON(body, ResumePost.class);
////	            if(resumePost !=null && resumePost.getOper() == 1){
////		            KeyedMessage<String, String> data = new KeyedMessage<String, String>(TOPIC, resumePost.getEntity());
////		            producer.send(data);
////		            logger.info(body);
////	            }
////	            messageNo ++;
////	        }
//		} catch (Exception e) {
//			logger.error("ERROR Occurs in function start() program will exit: ",e);
//		}finally{
//			System.exit(1);
//		}
//    }
//	
//	/**
//	 * @param args
//	 */
//	public static void main(String[] args) {
//		IMCProducer producer = IMCProducer.getInstance();
//		producer.start();
//	}
//}
